<?php

namespace Gnway\Util\Concurrent;

use SplQueue, Exception, RuntimeException;

/**
 * Class FixedPool
 *
 * @author zushaolong
 * @mail zushaolong@gmail.com
 */
class FixedPool extends AbstractPool
{

  /**
   * The pool core size
   *
   * @var int
   */
  protected int $corePoolSize;

  /**
   * Process scheduler
   *
   * @var Process|null
   */
  protected ?Process $scheduler = NULL;

  /**
   * FixedPool constructor.
   *
   * @param int $corePoolSize
   */
  public function __construct(int $corePoolSize = 4)
  {
    $this->corePoolSize = $corePoolSize;
  }

  /** @inheritDoc */
  protected function scheduler(): void
  {
    $this->scheduler = $this->newScheduler();

    $name = 'FixedPool[scheduler]';
    if (!empty($this->name)) {
      $name = $this->name . '[scheduler]';
    }
    $this->scheduler->setName($name);

    $this->scheduler->start();
  }

  /**
   * Make process scheduler
   *
   * @return Process
   */
  private function newScheduler(): Process
  {
    return new Process(new class($this->processes, $this->corePoolSize) implements Runnable {
      private SplQueue $tasks;
      private int $cores;

      public function __construct(array $tasks, int $cores)
      {
        $this->tasks = new SplQueue();
        $this->cores = $cores;

        foreach ($tasks as $task) {
          $this->tasks->enqueue($task);
        }
      }

      /**
       * Process exec the pool list
       *
       * @return void
       */
      public function run(): void
      {
        /** @var Process[] $started */
        $started = [];
        $tasks = $this->tasks;
        $process = Process::currentProcess();

        while (!$tasks->isEmpty() || $started) {
          if ($process->isInterrupted()) {
            $started && $this->waitTask($started, TRUE);
            break;
          }

          if (($tasks->isEmpty() && $started) || sizeof($started) >= $this->cores) {
            $this->waitTask($started, FALSE);
            usleep(10);
          }
          elseif (!$tasks->isEmpty()) {
            /** @var Process $task */
            $task = $tasks->dequeue();
            if ($task->isRunning()) {
              continue;
            }
            try {
              $task->start();
              $started[] = $task;
            } catch (Exception $ex) {
              echo 'Process startup: ' . $ex->getMessage() . PHP_EOL;
            }
          }
        }
      }

      /**
       * Waiting for tasks quit
       *
       * @param Process[] $started
       * @param bool $interrupted
       * @return void
       */
      private function waitTask(array &$started, bool $interrupted): void
      {
        foreach ($started as $idx => $task) {
          if ($task->isRunning()) {
            if (!$interrupted) {
              $task->wait(-1);
            } else {
              $task->shutdown(); //Block the task, until it exits
              unset($started[$idx]);
            }
          } else {
            unset($started[$idx]);
          }
        }
      }

    });
  }

  /** @inheritDoc */
  public function wait(int $timeout = 0): void
  {
    if ($this->scheduler === NULL) {
      throw new RuntimeException('No process scheduler');
    }

    $this->scheduler->wait($timeout);
  }

  /** @inheritDoc */
  public function shutdown(int $signal = SIGTERM): void
  {
    $this->scheduler->shutdown($signal);
  }
}